package com.shujia.tour

import com.shujia.utils.{Config, Geography, SSXRelation, SparkTool}
import org.apache.spark.sql.SQLContext


/**
  * 市游客计算
  * 1、出游时间大于3小时
  * 2、出游距离大于10KM
  * 关联停留表和用户画像表
  *
  */
object CityTouristJob extends SparkTool {
  /**
    * 子类实现此方法，实现具体的代码逻辑
    *
    */
  override def run(args: Array[String]): Unit = {

    if (args.length == 0) {
      log.error("请指定时间参数：")
      //抛出异常
      throw new RuntimeException("请指定时间参数：day_id")
    }
    val day_id = args(0)
    log.info(s"当前处理的时间分区为：$day_id")

    //读取停留表和用户画像表的数据

    val stayPointPath = Config.get("staypoint.path") + "day_id=" + day_id
    val stayPointRDD = sc.textFile(stayPointPath)

    log.info("停留表数据路径：" + stayPointPath)

    val usertagPath = Config.get("usertag.path") + "month_id=201805"
    val sQLContext = new SQLContext(sc)
    val userTagDF = sQLContext.read.parquet(usertagPath)

    log.info("用户画像表数据路径：" + usertagPath)


    //将停留表转换成kv格式，进行表关联
    val stayPointKVRDD = stayPointRDD.map(line => {
      val split = line.split("\t")
      val mdn = split(0)
      (mdn, line)
    })


    //取出常住地网格
    val userTagKVRDD = userTagDF.map(row => {
      val mdn = row.getAs[String]("mdn")
      //常住地网格编号
      val resiGridId = row.getAs[String]("resi_grid_id")
      val resiCountyId = row.getAs[String]("resi_county_id")
      (mdn, resiGridId + "-" + resiCountyId)
    })


    //区县获取城市的map集合
    val countyAndCityMap = SSXRelation.COUNTY_CITY
    val countyAndCityMapBro = sc.broadcast(countyAndCityMap)

    //观念停留表和用户画像表，计算每一个点的岛常住地网格的距离
    val mdnAndDistanceRDD = stayPointKVRDD.join(userTagKVRDD)
      .map(kv => {
        val mdn = kv._1
        val staypointLine = kv._2._1

        //常住地网格编号
        val resiGridId = kv._2._2.split("-")(0)
        //来源地区县
        val resiCountyId = kv._2._2.split("-")(1)

        //停留的网格编号
        val grId = staypointLine.split("\t")(3)

        //计算网格距离
        val distance = Geography.calculateLength(resiGridId.toLong, grId.toLong)

        // 获取目的地城市编码
        val countyId = staypointLine.split("\t")(4)
        //通过区县编号获取城市编号
        val cityId = countyAndCityMapBro.value.get(countyId)

        //获取每个点的停留时间
        val duration = staypointLine.split("\t")(5)


        val value = duration + "-" + distance + "-" + resiCountyId

        (mdn + "-" + cityId, value)
      })

    //计算每个人最大的出游距离，总的停留时间
    val mdnMaxDistanceRDD = mdnAndDistanceRDD
      //按照手机号和城市编号进行分组
      .groupByKey()
      .map(kv => {
        val mdn = kv._1.split("-")(0)
        val cityId = kv._1.split("-")(1)

        val iter = kv._2

        var maxDistance = Double.MinValue

        var stayTime = 0.0

        var resiCountyId = ""

        iter.foreach(line => {
          val duration = line.split("-")(0).toDouble
          val distance = line.split("-")(1).toDouble
          resiCountyId = line.split("-")(2)

          //最大出游距离
          if (distance > maxDistance) {
            maxDistance = distance
          }
          //计算总的停留时间
          stayTime += duration
        })

        (mdn, resiCountyId, cityId, stayTime, maxDistance)
      })

    //锅炉出游距离大于10km的游客
    val filterDistance = mdnMaxDistanceRDD.filter(line => {
      val stayTime = line._4
      val maxDistance = line._5

      /**
        * 1、出游时间大于3小时
        * 2、出游距离大于10KM
        */

      maxDistance > 10000 && stayTime > 180
    })

    val cityTouristRDD = filterDistance.map(line => {
      line._1 + "\t" + line._2 + "\t" + line._3 + "\t" + line._4 + "\t" + line._5
    })

    val cityTouristPath = Config.get("city.tourist.path") + "day_id=" + day_id

    log.info("数据输出路径：" + cityTouristPath)
    save(cityTouristRDD, cityTouristPath)
  }

  /**
    * 指定spark配置
    * conf.set(key,value)
    */
  override def init(): Unit = {
    conf.set("spark.files.fetchTimeout", "1000s")
  }
}
